package spark_core



import org.apache.spark.sql.{ForeachWriter, Row}

import java.sql.{Connection, DriverManager, PreparedStatement}

class JdbcUtil(url:String,username:String,password:String) extends ForeachWriter[Row]{
  val sql:String="insert into user(userid,username,usersex) value(?,?,?)"
  var connection:Connection= _
  var driver:String = "com.mysql.cj.jdbc.Driver"
  var statement:PreparedStatement= _

  override def open(partitionId: Long, epochId: Long): Boolean ={
    connection=DriverManager.getConnection(url,username,password)
    statement=connection.prepareStatement(sql)
    true
  }

  override def process(value: Row): Unit ={
    println(value)
    statement.setString(1,value.get(0).toString)
    statement.setString(2,value.get(1).toString)
    statement.setString(3,value.get(2).toString)
    statement.executeUpdate()
  }

  override def close(errorOrNull: Throwable): Unit ={
    connection.close()
  }
}
